package com.eshop.storm.bolt;

import com.alibaba.fastjson.JSONObject;
import com.eshop.storm.cons.ProjectCons;
import com.eshop.storm.enums.ZkLock;
import com.eshop.storm.utils.DateUtils;
import com.eshop.storm.utils.HttpClientUtils;
import com.eshop.storm.zk.ZooKeeperSession;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.NameValuePair;
import org.apache.http.client.utils.URLEncodedUtils;
import org.apache.http.message.BasicNameValuePair;
import org.apache.http.protocol.HTTP;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.trident.util.LRUMap;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.utils.Utils;

import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author: Xukai
 * @description:
 * @createDate: 2018/8/2 01:02
 * @modified By:
 */
@Slf4j
public class ProductCountBolt extends BaseBasicBolt {

    private static final long serialVersionUID = 2482624042843007601L;

    private ZooKeeperSession zooKeeperSession;

    private Integer taskId;

    private LRUMap<Long, Long> productCountMap = new LRUMap<>(1000);

    @Override
    public void prepare(Map stormConf, TopologyContext context) {
        zooKeeperSession = ZooKeeperSession.getInstance();
        taskId = context.getThisTaskId();
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("product-count-" + taskId).build();
        ThreadPoolExecutor threadExecutor = new ThreadPoolExecutor(
                2,
                2,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(),
                namedThreadFactory,
                new ThreadPoolExecutor.AbortPolicy());
        threadExecutor.submit(new ProductCountThread());
        threadExecutor.submit(new HotProductFindThread());
        // 1、将自己的taskid写入一个zookeeper node中，形成taskid的列表
        // 2、然后每次都将自己的热门商品列表，写入自己的taskid对应的zookeeper节点
        // 3、然后这样的话，并行的预热程序才能从第一步中知道，有哪些taskid
        // 4、然后并行预热程序根据每个taskid去获取一个锁，然后再从对应的znode中拿到热门商品列表
        initTaskId(taskId);
    }

    private void initTaskId(Integer taskId) {
        // ProductCountBolt所有的task启动的时候，都会将自己的taskid写到同一个node的值中
        // 格式就是逗号分隔，拼接成一个列表
        // 111,211,355
        zooKeeperSession.acquireDistributedLock(ZkLock.TASK_ID_LIST_LOCK.getType(), null);
        String taskIdList = zooKeeperSession.getNodeData(ZkLock.TASK_ID_LIST.getType(), null);
        Boolean existFlag = false;
        if (StringUtils.isNotEmpty(taskIdList)) {
            for (String splitTaskId : taskIdList.split(",")) {
                if (splitTaskId.equals(String.valueOf(taskId))) {
                    existFlag = true;
                }
            }
            if (!existFlag) {
                taskIdList = "," + taskId;
            }
        } else {
            taskIdList = "" + taskId;
        }
        zooKeeperSession.setNodeDate(ZkLock.TASK_ID_LIST.getType(), null, taskIdList);
        zooKeeperSession.releaseDistributedLock(ZkLock.TASK_ID_LIST_LOCK.getType(), null);
    }

    /**
     * 热门商品统计线程
     */
    private class ProductCountThread implements Runnable {

        private TreeMap<Long, Long> topnProductCountMap = new TreeMap<>((a, b) -> (int) (b - a));

        @Override
        public void run() {
            while (true) {
                try {
                    if (productCountMap.size() == 0) {
                        Utils.sleep(100);
                        continue;
                    }
                    topnProductCountMap.clear();
                    for (Map.Entry<Long, Long> entry : productCountMap.entrySet()) {
                        Long productId = entry.getKey();
                        Long count = entry.getValue();
                        topnProductCountMap.put(count, productId);
                        if (topnProductCountMap.size() > ProjectCons.TOPN_HOT_PRODUCT_SIZE) {
                            topnProductCountMap.remove(topnProductCountMap.lastKey());
                        }
                    }
                    log.info("计算得到热门商品:{}", topnProductCountMap);
                    // 将topN的Map<count, productId>存入zk
                    String topNProductJSON = JSONObject.toJSONString(topnProductCountMap);
                    zooKeeperSession.setNodeDate(ZkLock.TASK_HOT_PRODUCT.getType(), (long) taskId, topNProductJSON);
                    // 将taskIdStatus的节点值设为空，表示该task对应的数据可以被执行预热
                    zooKeeperSession.setNodeDate(ZkLock.TASK_ID_STATUS.getType(), Long.valueOf(taskId), "");
                    Utils.sleep(ProjectCons.PRODUCT_COUNT_INTERVAL);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 热点数据统计线程
     */
    public class HotProductFindThread implements Runnable {

        private TreeMap<Long, Long> hotProductFindMap = new TreeMap<>((a, b) -> (int) (b - a));
        List<Long> hotProductIdList = new ArrayList<>();
        List<Long> lastestHotProductIdList = new ArrayList<>();
        Map<Long, Long> hotProductIdBirthTime = new HashMap<>();

        @Override
        public void run() {
            while (true) {
                try {
                    // 1、将LRUMap中的数据按照访问次数，进行全局的排序
                    // 2、计算95%的商品的访问次数的平均值
                    // 3、遍历排序后的商品访问次数，从最大的开始，如果某个商品比如它的访问量是平均值的10倍，就认为是缓存的热点
                    if (productCountMap.size() == 0) {
                        Utils.sleep(100);
                        continue;
                    }
                    hotProductFindMap.clear();
                    hotProductIdList.clear();
                    // 1、先做全局的排序
                    for (Map.Entry<Long, Long> entry : productCountMap.entrySet()) {
                        Long productId = entry.getKey();
                        Long count = entry.getValue();
                        hotProductFindMap.put(count, productId);
                    }
                    // 2、计算出95%的商品的访问次数的平均值
                    Long sampleCount = 0L;
                    int sampleSize = (int) Math.floor(productCountMap.size() * ProjectCons.HOT_PRODUCT_SAMPLE_RATION);
                    if (sampleSize == 0) {
                        sampleSize = 1;
                    }
                    int index = 1;
                    for (Long count : hotProductFindMap.keySet()) {
                        if (index > (productCountMap.size() - sampleSize)) {
                            sampleCount += count;
                        }
                        index++;
                    }
                    // 3、从第一个元素开始遍历，判断是否是平均值的10倍
                    Long thresholdCount = sampleCount / sampleSize * ProjectCons.HOT_PRODUCT_COUNT_THRESHOLD;
                    Long productId;
                    for (Map.Entry<Long, Long> entry : hotProductFindMap.entrySet()) {
                        if (entry.getKey() > thresholdCount) {
                            productId = entry.getValue();
                            hotProductIdList.add(productId);
                            // 如果上一次的热点数据列表已经不包含这个ID，说明是新热点ID，或者热点ID的诞生时间已经超时
                            // 就需要重新设置分发层和应用层，通过此方法筛选减小缓存服务的压力
                            if (!lastestHotProductIdList.contains(productId) || DateUtils.before(hotProductIdBirthTime.get(productId) + ProjectCons.HOT_PRODUCT_CACHE_SURVIVAL_TIME, System.currentTimeMillis())) {
                                // 重新设置热点ID的诞生时间
                                hotProductIdBirthTime.put(productId, System.currentTimeMillis());
                                // 将缓存热点商品ID反向推送到流量分发的nginx上
                                String distributeNginxURL = ProjectCons.DISTRIBUTE_NGINX_SET_HOT_PRODUCT_BASE_URL + productId;
                                HttpClientUtils.sendGetRequest(distributeNginxURL);
                                // 将缓存热点商品对应的完整缓存数据，先发送请求到缓存服务去获取，再反向推送到所有的后端应用nginx服务器上
                                String productDataUrl = ProjectCons.PRODUCT_SERVICE_BASE_URL + productId;
                                String response = HttpClientUtils.sendGetRequest(productDataUrl);
                                // 对Json数据做URL编码，否则会报错
                                List<NameValuePair> params = new ArrayList<>();
                                params.add(new BasicNameValuePair("productInfo", response));
                                String productInfo = URLEncodedUtils.format(params, HTTP.UTF_8);
                                String[] appNginxURLs = new String[]{
                                        ProjectCons.APP_NGINX_01_HOT_PRODUCT_BASE_URL[0] + productId + ProjectCons.APP_NGINX_01_HOT_PRODUCT_BASE_URL[1] + productInfo,
                                        ProjectCons.APP_NGINX_02_HOT_PRODUCT_BASE_URL[0] + productId + ProjectCons.APP_NGINX_02_HOT_PRODUCT_BASE_URL[1] + productInfo
                                };
                                for (String appNginxURL : appNginxURLs) {
                                    HttpClientUtils.sendGetRequest(appNginxURL);
                                }
                            }
                        } else {
                            break;
                        }
                    }
                    log.info("当前热点数据商品ID:{}", hotProductIdList);
                    // 4、实时感知热点数据的消失
                    if (lastestHotProductIdList.size() != 0) {
                        for (Long lastestHotProductId : lastestHotProductIdList) {
                            // 如果当前热点列表不包含上一次热点列表的数据，说明这个数据已经不是热点数据
                            // 就需要通知分发Nginx取消该数据的热点身份
                            // 同时删除hotProductIdBirthTime中对应的诞生时间
                            if (!hotProductIdList.contains(lastestHotProductId)) {
                                String distributeNginxCanceLURL = ProjectCons.DISTRIBUTE_NGINX_CANCEL_HOT_PRODUCT_BASE_URL + lastestHotProductId;
                                HttpClientUtils.sendGetRequest(distributeNginxCanceLURL);
                                hotProductIdBirthTime.remove(lastestHotProductId);
                            }
                        }
                    }
                    lastestHotProductIdList.clear();
                    for (Long hotProductId : hotProductIdList) {
                        lastestHotProductIdList.add(hotProductId);
                    }
                    Utils.sleep(ProjectCons.HOT_PRODUCT_FIND_INTERVAL);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    @Override
    public void execute(Tuple input, BasicOutputCollector collector) {
        Long productId = input.getLongByField("productId");
        Long count = productCountMap.get(productId);
        if (count == null) {
            count = 0L;
        }
        count++;
        productCountMap.put(productId, count);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }
}
